package queue

import (
	"sync"
	"time"
)

type Receiver struct {
	mu             sync.Mutex
	bs             *Broadcasting
	invokes        *Invoke
	handler        func(string) //回调方法
	index          int32        //消耗到的下标
	concurrencyNum int32        //当前异步回调并行数量
	maxConcurrency int32        //最大异步回调并行数量
	in             chan *Invoke
}

func (rc *Receiver) getInvoke() *Invoke {
	if rc.maxConcurrency == 0 {
		panic("maxConcurrency is zero")
	}

	var in *Invoke
	rc.mu.Lock()
	if rc.invokes != nil {
		in = rc.invokes
		rc.invokes = rc.invokes.next
	}

	if in == nil {
		if rc.concurrencyNum < rc.maxConcurrency {
			rc.concurrencyNum++
			in = rc.generatorInvoke()
		}
	}
	rc.mu.Unlock()
	if in == nil {
		<-time.After(time.Millisecond)
		return rc.getInvoke()
	}
	return in
}

func (rc *Receiver) generatorInvoke() *Invoke {
	return &Invoke{
		rc: rc,
	}
}

func (rc *Receiver) run() {
	for {
		select {
		case in := <-rc.in:
			rc.mu.Lock()
			in.next = rc.invokes
			rc.invokes = in
			rc.mu.Unlock()
		default:
			l := len(rc.bs.msg) //可以不锁，有延迟没关系
			index := int(rc.index)
			if index < l {
				rc.index++
				go func() {
					in := rc.getInvoke()
					in.invoke(rc.bs.msg[index].data)
					in.rc.in <- in
				}()
			}else{
				<-time.After(time.Millisecond)
			}
		}
	}
}
